package com.lzwk.app.ods.adb;

import com.lzwk.app.function.GetJsonDouble;
import com.lzwk.common.test.TestConstant;
import org.apache.calcite.schema.StreamableTable;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/*
 *@Author:yeaser
 *@Date:2022/4/2-0:09
 *Decription:com.lzwk.app.ods.adb
 *@Version:1.8
 */
public class AdbActivityTest {
    public static void main(String[] args) {
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        tenv.createTemporaryFunction("GetJsonDouble", GetJsonDouble.class);

        tenv.executeSql("create table activity_term(" +
                "  `id` INT, " +
                "  `division_activity_id` INT, " +
                "  `parent_id` INT, " +
                "  `grade` INT, " +
                "  `content_sync_term` INT, " +
                "  `wrapper_sync_term` INT, " +
                "  `rank` INT, " +
                "  `template_id` INT, " +
                "  `template_type` STRING, " +
                "  `start_time` TIMESTAMP(0), " +
                "  `end_time` TIMESTAMP(0), " +
                "  `class_limit` INT, " +
                "  `class_student_limit` INT, " +
                "  `student_cnt` INT, " +
                "  `state` STRING, " +
                "  `create_time` TIMESTAMP(0), " +
                "  `update_time` TIMESTAMP(0), " +
                "  `data_end_time` TIMESTAMP(0), " +
                "  `promote_start_time` TIMESTAMP(0), " +
                "  `remark` STRING, " +
                " `proc_time` AS PROCTIME(), " +
                " WATERMARK FOR create_time AS create_time - INTERVAL '1' SECOND, " +
                "  PRIMARY KEY (`id`) NOT ENFORCED" +
                " )with(" +
                "   'connector'='mysql-cdc', " +
                "   'hostname'='" + TestConstant.TESTHOST + "', " +
                "   'port'='" + TestConstant.TESTPORT + "', " +
                "   'username'='" + TestConstant.TESTUSERNAME + "', " +
                "   'password'='" + TestConstant.TESTPASSWORD + "', " +
                "   'database-name'='" + TestConstant.TESTADBDB + "', " +
                "   'table-name'='" + TestConstant.TESTADBJDBCDIVISIONACTIVITYTERM + "', " +
                "   'server-time-zone'='Asia/Shanghai', " +
                "   'debezium.snapshot.locking.mode' = 'none', " +
                "   'scan.incremental.snapshot.chunk.size' = '80000' " +
                ")");

        tenv.sqlQuery(" select id," +
                "ifnull(GetJsonDouble(remark, 'input_cost'), 0.00) as input_cost, " +
                "ifnull(GetJsonDouble(remark, 'target_roi'), 0.00) as target_roi, " +
                "remark,create_time from activity_term " +
                //"where remark is null " +
                "").execute().print();



    }
}
